package cn.doitedu.data_sync;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/5/5
 * @Desc: 学大数据，上多易教育
 *   用户注册信息表，从业务库，同步到 hbase
 **/
public class Job01_UmsMember_Hbase {


    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///D:/flink/ck");


        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 建表，映射mysql中的用户注册信息表
        tenv.executeSql(
                " create table ums_member_mysqlcdc(          "+
                        "   id bigint                                "+
                        "   ,member_level_id bigint                  "+
                        "   ,username string                         "+
                        "   ,password string                         "+
                        "   ,nickname string                         "+
                        "   ,phone string                            "+
                        "   ,status int                              "+
                        "   ,create_time timestamp(3)                "+
                        "   ,icon string                             "+
                        "   ,gender int                              "+
                        "   ,birthday date                           "+
                        "   ,city string                             "+
                        "   ,job string                              "+
                        "   ,personalized_signature string           "+
                        "   ,source_type int                         "+
                        "   ,integration int                         "+
                        "   ,growth int                              "+
                        "   ,luckey_count int                        "+
                        "   ,history_integration int                 "+
                        "   ,modify_time timestamp(3)                "+
                        "   ,primary key(id)  not enforced           "+
                        " ) with (                                   "+
                        " 'connector' = 'mysql-cdc',                 "+
                        " 'hostname' = 'doitedu',                    "+
                        " 'port' = '3306',                           "+
                        " 'username' = 'root',                       "+
                        " 'password' = 'root',                       "+
                        " 'database-name' = 'realtimedw',            "+
                        " 'table-name' = 'ums_member'                "+
                        " )                                          "

        );


        // 建表，映射hbase中的目标表

        tenv.executeSql(
                        " create table ums_member_hbase(          "+
                        "   username string                       "+
                        "   ,f ROW<                               "+
                        "    id bigint                             "+
                        "   ,member_level_id bigint               "+
                        "   ,nickname string                      "+
                        "   ,phone string                         "+
                        "   ,status int                           "+
                        "   ,create_time timestamp(3)             "+
                        "   ,icon string                          "+
                        "   ,gender int                           "+
                        "   ,birthday date                        "+
                        "   ,city string                          "+
                        "   ,job string                           "+
                        "   ,personalized_signature string        "+
                        "   ,source_type int                      "+
                        "   ,integration int                      "+
                        "   ,growth int                           "+
                        "   ,luckey_count int                     "+
                        "   ,history_integration int              "+
                        "   ,modify_time timestamp(3)   >         "+
                        " ) WITH (                                "+
                        "  'connector' = 'hbase-2.2',             "+
                        "  'table-name' = 'dim_user_info',        "+
                        "  'zookeeper.quorum' = 'doitedu:2181',    "+
                        "  'lookup.cache' = 'PARTIAL',                          " +
                        "  'lookup.partial-cache.max-rows' = '100000',          " +
                        "  'lookup.partial-cache.expire-after-access' = '1 h',  " +
                        "  'lookup.partial-cache.cache-missing-key' = 'true'  " +
                        " )                                                      "
        );





        tenv.executeSql(
                        " INSERT INTO ums_member_hbase                                                         "+
                        " SELECT                                                                               "+
                       // "  substring(MD5(cast(id as string)),0,4) || lpad(cast(id as string),8,'0')  as  id  "+
                        "  username                                                                            "+
                        "  ,row(id,member_level_id                                                                "+
                        "  ,nickname                                                                           "+
                        "  ,phone                                                                              "+
                        "  ,status                                                                             "+
                        "  ,create_time                                                                        "+
                        "  ,icon                                                                               "+
                        "  ,gender                                                                             "+
                        "  ,birthday                                                                           "+
                        "  ,city                                                                               "+
                        "  ,job                                                                                "+
                        "  ,personalized_signature                                                             "+
                        "  ,source_type                                                                        "+
                        "  ,integration                                                                        "+
                        "  ,growth                                                                             "+
                        "  ,luckey_count                                                                       "+
                        "  ,history_integration                                                                "+
                        "  ,modify_time )  as f                                                                "+
                        "                                                                                      "+
                        " from ums_member_mysqlcdc                                                             "

        );



    }



}
